package com.navinfo.tripanalysis.offline;

import com.navinfo.tripanalysis.common.arithmetic.common.OuterEventData;
import com.navinfo.tripanalysis.common.arithmetic.common.OuterStatisticData;
import com.navinfo.tripanalysis.common.util.CommonUtils;
import com.navinfo.tripanalysis.offline.pojo.LoadDataParam;
import com.navinfo.tripanalysis.common.pojo.RunningTypeEnum;
import com.navinfo.tripanalysis.offline.service.*;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Options;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Properties;

import static com.navinfo.tripanalysis.offline.util.SparkAppUtils.getSparkBuilder;

/**
 * 驾驶行为分析 将hive中的行程统计转存到mongo及事件数据转存到hbase中
 * @author 沈东生
 */
public class TripEtlApplication extends TripAnalysisApplication {
    private static final Logger logger = LoggerFactory.getLogger(TripEtlApplication.class);
    private static final String savePrefix = "etl.";

    public static void main(String[] args) throws Exception {
        logger.error("开始离线驾驶行为分析ETL工作，args:{}", Arrays.toString(args));

        //加载配置
        Options options = setCmdOptions();
        CommandLine cmd = parseCmdOptions(options, args);
        Properties config = loadProps(cmd);
        parseDays(args, cmd);

        String runType = config.getProperty("system.running.type");
        if(StringUtils.isEmpty(runType)){
            logger.error("请配置系统运行类型，1东风，2青汽");
            System.exit(1);
        }

        runningTypeEnum = RunningTypeEnum.valueOf(Integer.valueOf(runType));
        logger.error("离线驾驶行为分析ETL工作，系统运行类型{},{}", runningTypeEnum.getType(), runningTypeEnum.getDesc());

        //初始化SparkSession
        SparkSession spark = getSparkBuilder(config).getOrCreate();
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());

        //创建相关服务
        ServiceFactory factory = new ServiceFactory(config, runningTypeEnum);
        LoadTripStatisticService loadTripStatisticService = factory.createLoadTripService();
        LoadTripEventService loadTripEventService = factory.createLoadEventService();
        List<SaveTripStatisticService> saveTripStatisticServices = factory.createSaveTripServices(savePrefix);
        List<SaveTripEventService> saveTripEventServices = factory.createSaveEventServices(savePrefix);

        //加载回放特定tid的列表
        LoadDataParam param = new LoadDataParam();
        param.setTidList(loadTidList(cmd));

        //按天进行行程和事件的导出
        for (long d = start; d < end; d += DAY_MS) {
            param.setDay(new Date(d));

            if (CommonUtils.isCollectionNotEmpty(saveTripStatisticServices)) {
                JavaPairRDD<Long, List<OuterStatisticData>> tripRdd = loadTripStatisticService.load(spark, param);
                for (SaveTripStatisticService saveTripStatisticService : saveTripStatisticServices) {
                    saveTripStatisticService.save(spark, jsc, tripRdd, d);
                }
            } else {
                logger.error("saveTripServices is empty");
            }

            if (CommonUtils.isCollectionNotEmpty(saveTripEventServices)) {
                JavaPairRDD<Long, List<OuterEventData>> eventRdd = loadTripEventService.load(spark, param);
                for (SaveTripEventService saveTripEventService : saveTripEventServices) {
                    saveTripEventService.save(spark, jsc, eventRdd, d);
                }
            } else {
                logger.error("saveEventServices is empty");
            }

        }

        logger.error("完成离线驾驶行为分析ETL工作......");
        jsc.close();
        System.exit(0);
    }
}
